package com.codejiwei.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/6/9 16:59
 */
public class Flink_SQL_Kafka2Print {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(10);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        tableEnv.executeSql("create table source_table (id string, ts timestamp(3), vc int) with("
                + "'connector' = 'kafka',"
                + "'topic' = 'drity_data_source',"
                + "'properties.bootstrap.servers' = '192.168.102.154:29092',"
                + "'properties.group.id' = 'jiwei01',"
                + "'scan.startup.mode' = 'latest-offset',"
                + "'format' = 'json'"
                + ")");

        tableEnv.executeSql("CREATE TABLE sink_table WITH ('connector' = 'print')\n" +
                "LIKE source_table (EXCLUDING ALL)");

        tableEnv.executeSql("insert into sink_table select * from source_table");



    }
}
